package com.ylw.kafka;


import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;


/**
 * Broker：Kafka 集群包含一个或多个服务器，这种服务器被称为 broker。
 * Topic：每条发布到 Kafka 集群的消息都有一个类别，这个类别被称为 Topic。（物理上不同 Topic 的消息分开存储，逻辑上一个 Topic 的消息虽然保存于一个或多个 broker 上，但用户只需指定消息的 Topic 即可生产或消费数据而不必关心数据存于何处）。
 * Partition：Partition 是物理上的概念，每个 Topic 包含一个或多个 Partition。
 * Producer：负责发布消息到 Kafka broker。
 * Consumer：消息消费者，向 Kafka broker 读取消息的客户端。
 * Consumer Group：每个 Consumer 属于一个特定的 Consumer Group（可为每个 Consumer 指定 group name，若不指定 group name 则属于默认的 group）。
 */
public class KafkaUtils {

    private final static Logger LOGGER = LoggerFactory.getLogger(KafkaUtils.class);

    private static Producer<String, String> producer;
    private static Consumer<String, String> consumer;

    private KafkaUtils() {

    }

    /**
     * 生产者，注意kafka生产者不能够从代码上生成主题，只有在服务器上用命令生成
     */
    static {
        Properties props = new Properties();
        props.put("bootstrap.servers", Constants.BOOTSTRAP_SERVERS);//服务器ip:端口号，集群用逗号分隔
        props.put("acks", "all");
        props.put("retries", 3);//重试次数
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);

//        props.put("producer.type", "sync");
//        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");


        producer = new KafkaProducer(props);
    }

    /**
     * 消费者
     */
    static {
        Properties props = new Properties();
        props.put("bootstrap.servers", Constants.BOOTSTRAP_SERVERS);//服务器ip:端口号，集群用逗号分隔
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");

//        props.put("key.serializer", "kafka.serializer.StringDeserializer");
//        props.put("value.serializer", "kafka.serializer.StringDeserializer");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer(props);
        consumer.subscribe(Arrays.asList(Constants.TOPIC_NAME));
    }

    /**
     * 发送对象消息 至kafka上,调用json转化为json字符串，应为kafka存储的是String。
     * @param msg
     */
    public static void sendMsgToKafka(String msg) throws ExecutionException, InterruptedException {
        System.out.println("### send start");
        Future<RecordMetadata> result =  producer.send(new ProducerRecord<String, String>(Constants.TOPIC_NAME, String.valueOf(new Date().getTime()),
                msg));
        System.out.println(result.get());
        RecordMetadata ss  = result.get();
        System.out.println("## topic "+ss.topic());
        System.out.println("## ss "+ss.toString());
        System.out.println("#  send end");
    }


    /**
     * 从kafka上接收对象消息，将json字符串转化为对象，便于获取消息的时候可以使用get方法获取。
     */
    public static void getMsgFromKafka(){
        while(true){
            ConsumerRecords<String, String> records = KafkaUtils.getKafkaConsumer().poll(100);
            if (records.count() > 0) {
                for (ConsumerRecord<String, String> record : records) {
                    JSONObject jsonAlarmMsg = JSON.parseObject(record.value());
                    if(jsonAlarmMsg.getString("traceId").equals("cc93aa36dd8046cbb6e7338c81e04908")){
                        System.out.println("#### "+ jsonAlarmMsg);
                        return;
                    }
                    LOGGER.info("从kafka接收到的消息是：" + jsonAlarmMsg.toString());
                }
            }
        }
    }

    public static Consumer<String, String> getKafkaConsumer() {
        return consumer;
    }

    public static void closeKafkaProducer() {
        producer.close();
    }

    public static void closeKafkaConsumer() {
        consumer.close();
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        sendMsgToKafka("{\"traceId\":\"bd7a977555f6b982bd7a977555f6b988\",\"name\":\"C-transferOrderToCSB\",\"id\":\"bd7a977555f8c777\",\"parentId\":\" bd7a977555f7c888\",\"timestamp\":1486989554481000,\"duration\":10000,\"annotations\":[{\"timestamp\":1486989554481000,\"value\":\"cs\",\"endpoint\":{\"serviceName\":\"1001.4G 新装订单转集团CSB\",\"ip\":\"192.168.1.4\",\"port\":9500}},{\"timestamp\":1486989554581000,\"value\":\"cr\",\"endpoint\":{\"serviceName\":\"1001. 4G 新装订单转集团CSB\",\"ip\":\"192.168.1.4\",\"port\":9500}}],\"binaryAnnotations\":[{\"key\":\"bz.resultCode\",\"value\":\"0\",\"endpoint\":{\"serviceName\":\"1001. 4G 新装订单转集团CSB\",\"ip\":\"192.168.1.4\",\"port\":9500}}]}");

        System.out.println("-----------------------");

        getMsgFromKafka();

    }


}



class  Constants{
    static String BOOTSTRAP_SERVERS = "";
    static String TOPIC_NAME = "";

}

